﻿//Description:捕捉和推送类-数据变更记录
//Copyright (c) : 通力凯顿（北京）系统集成有限公司
//Writer:Wangjh
//create Date:2023-8-9
//Rewriter:
//Rewrite Date:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using DatabaseAccess.DBUtility;
using BusinessRule.Business;
using System.Collections;
using BusinessRule.Common;
using System.Collections.Concurrent; 
using System.Configuration;
using System.Net;
using System.Data.OracleClient;
using Newtonsoft.Json;
using System.IO;
using System.Threading.Tasks;
using BusinessRule.DTO;

namespace OracleTableSync
{
   public  class CaptureAndPush
    {
       SyncTableStructure sts = new SyncTableStructure();
       SyncTableAllData sta = new SyncTableAllData();
       private static readonly String orgDb = ConfigurationManager.AppSettings["orgDb"].ToString();
       private static readonly String toDb = ConfigurationManager.AppSettings["toDb"].ToString();
       private static readonly int maxParallel = Environment.ProcessorCount;//最大并行数
       private const int maxData=2000;//一个json文件容纳的最大数据量，暂定2000
       private List<Task> allTaskList ;//异步的任务

       /// <summary>
       /// 执行入口
       /// </summary>
       public  void Excute() {
           LogHelp.SetConfig();
           allTaskList = new List<Task>();
           //处理需要推送的表结构
           //sts.DealSyncStructData();
           //处理推送数据
           DealSyncData();
           Task.WaitAll(allTaskList.ToArray());//等待异步的逻辑
        }
          

       /// <summary>
       /// 获取已到同步时间的表信息
       /// </summary>
       /// <returns></returns>
       DataTable GetSyncTableData() {
           String strSql = @"select t.* from sync_TableInfo t where t.nextsynctime<=sysdate";

           DataTable dt = DatabaseHelper.Query_ReturnDatTable(strSql);
           return dt;
       }

        /// <summary>
        /// 获取需要推送的数据,分页查询一次默认2000条
        /// </summary>
        /// <returns></returns>
       uMESProcessResult GetSyncData(String tableName,int i) {


           String strSql = @"select sr.id,sr.tablename,sr.UniqueColName,sr.oldvalue,sr.newvalue,
to_char(sr.createtime,'YYYY-MM-DD HH24:MI:SS.FF3') createtime,TO_CHAR(sr.lastchangetime,'YYYY-MM-DD HH24:MI:SS.FF3') lastchangetime,
sr.sync,sr.operatetype,sr.ishandle,sr.uniquecolvalue from Sync_TableChangeRecord sr

where sr.ISHANDLE=0 and sr.tablename='{0}'
order by sr.lastchangetime";


           uMESPagingDataDTO paraPage = new uMESPagingDataDTO();
           paraPage.CurrentPageIndex =i;
           paraPage.PageSize = maxData;
           paraPage.strSQL = String.Format(strSql, tableName);
           return  DatabaseHelper.RunProcedureForPage(paraPage);
             
       }
        
       /// <summary>
       /// 处理需要推送的数据,一次处理同一张表的数据 
       /// </summary>
       /// <param name="dt"></param>
       void DealSyncData(DataRow row) {

           int allCount = int.Parse(DatabaseHelper.GetSingle(String.Format("SELECT COUNT(1) FROM Sync_TableChangeRecord WHERE ISHANDLE=0 AND TABLENAME='{0}'",row["TABLENAME"].ToString ())).ToString());
           if (allCount == 0)
               return;
           double pageCount = Math.Ceiling((double)allCount / maxData);//获取总页数

           for (int j = 0; j < pageCount; j++)
           {
               #region 单页数据处理
               uMESProcessResult re = GetSyncData(row["tablename"].ToString (),1);
               if (re.DBTable == null)
                   continue;
               if (re.DBTable.Rows.Count == 0)
                   continue;
                DataTable dt = re.DBTable;

                string tableName = dt.Rows[0]["tablename"].ToString();//本次处理的表
                List<ExcuteEntity> exs = new List<ExcuteEntity>();
                List<ExcuteEntity> exsAsync = new List<ExcuteEntity>();//异步执行的sql
                String pushID = Guid.NewGuid().ToString();//一次推送的id 

                String receiveSql = @"select TO_CHAR(sr.lastchangetime,'YYYY-MM-DD HH24:MI:SS.FF3') lastchangetime,sr.oldvalue,sr.newvalue,sr.OperateType  from Sync_ReceiveRecord sr 
where sr.TableName='{0}' and sr.UniqueColValue='{1}' and sr.receiveid in
(
select receiveid from (
select row_number() over(order by sr.lastchangetime desc) rn,
sr.receiveid
from Sync_ReceiveRecord sr 
where sr.TableName='{0}' and sr.UniqueColValue='{1}' 
) where rn=1
)";

                int num = 0;
                List<Dictionary<String, Object>> jsonFile = new List<Dictionary<String, Object>>();
                for (int i = 0; i < dt.Rows.Count; i++)
                {
                    DataRow r = dt.Rows[i];
                    int sync = 1;//应该同步
                    String validateSql = string.Format(receiveSql, r["tablename"].ToString(), r["UniqueColValue"].ToString());

                    DataTable validateDt = DatabaseHelper.Query_ReturnDatTable(validateSql);//此条记录最近的接收处理情况,只有两端更新同一张表时才会有此验证

                    if (validateDt.Rows.Count > 0)
                    {
                        foreach(DataRow validateRow in validateDt.Rows){
                            if (r["OperateType"].ToString() == "3" && validateRow["OperateType"].ToString() == "3")//最近删除过,不应该推送//更新状态
                            {
                                sync = 0;
                                break;
                            }
                            if (r["oldvalue"].ToString() == validateRow["oldvalue"].ToString())//旧值相同
                            {
                                if (CommonHelper.checkEqualsNewValues(r["newvalue"].ToString(), validateRow["newvalue"].ToString()))
                                {
                                    sync = 0;
                                    break;
                                }                                
                            } 
                        }
                         

                    }

                    //更新状态
                    exs.Add(new ExcuteEntity()
                    {
                        TableName = "Sync_TableChangeRecord",
                        ExcuteType = ExcuteEntity.ExcuteTypeEnum.update,
                        ExcuteFileds = new List<FieldEntity>() { new FieldEntity("Sync", sync, FieldEntity.FieldTypeEnum.Numer),
                        new FieldEntity("IsHandle", 1, FieldEntity.FieldTypeEnum.Numer),
                        new FieldEntity("LastChangeTime",DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp)},
                        WhereFileds = new List<FieldEntity>() { new FieldEntity("Id", r["Id"].ToString(), FieldEntity.FieldTypeEnum.Str) }
                    });

                    if (sync == 1)
                    {
                        num++;
                        //插入推送表
                        exsAsync.Add(new ExcuteEntity()
                        {
                            TableName = "Sync_PushRecord",
                            ExcuteType = ExcuteEntity.ExcuteTypeEnum.insert,
                            ExcuteFileds = new List<FieldEntity>() { new FieldEntity("id", Guid.NewGuid().ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("PushID", pushID, FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("TableChangeRecordID", r["Id"].ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("TableName", r["TableName"].ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("OldValue", r["OldValue"].ToString(), FieldEntity.FieldTypeEnum.Clob),
                           new FieldEntity("NewValue", r["NewValue"].ToString(), FieldEntity.FieldTypeEnum.Clob),
                           new FieldEntity("OperateType", r["operatetype"], FieldEntity.FieldTypeEnum.Numer),
                           new FieldEntity("orgDb", orgDb, FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("toDb", toDb, FieldEntity.FieldTypeEnum.Str), 
                           new FieldEntity("CreateTime", DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp),
                           new FieldEntity("LastChangeTime", DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp),
                           new FieldEntity("UniqueColValue", r["UniqueColValue"].ToString (), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("UniqueColName", r["UniqueColName"].ToString (), FieldEntity.FieldTypeEnum.Str)
                       }
                        });
                        //存入json文件
                        Dictionary<String, Object> eachJson = new Dictionary<String, Object>();
                        eachJson.Add("type", "syncData");
                        eachJson.Add("pushID", pushID);
                        eachJson.Add("tableChangeRecordID", r["id"].ToString());
                        eachJson.Add("oldValue", r["oldValue"].ToString());
                        eachJson.Add("newValue", r["NewValue"].ToString());
                        eachJson.Add("tableName", r["TableName"].ToString());
                        eachJson.Add("uniqueColName", r["UniqueColName"].ToString());
                        eachJson.Add("uniqueColValue", r["UniqueColValue"].ToString());
                        eachJson.Add("operateType", r["operateType"].ToString());
                        eachJson.Add("orgDb", orgDb);
                        eachJson.Add("toDb", toDb);

                        jsonFile.Add(eachJson);
                    }
                     
                }
                if (exs.Count > 0)
                { 
                    DatabaseHelper.ExecuteDataByEntitys(exs); 
                }
                if (exsAsync.Count > 0) {
                    allTaskList.Add(Task.Factory.StartNew(() =>{
                        try {
                            DatabaseHelper.ExecuteDataByEntitys(exsAsync);
                        }
                        catch (Exception ex)
                        {
                            LogHelp.WriteError("捕捉和推送-存入推送记录异常", ex);                        
                        }
                    }));
                }
                if (jsonFile.Count > 0)
                {
                    String fileName = CommonHelper.writeJsonToFile(jsonFile, "SyncData_" + jsonFile[0]["tableName"].ToString());
                    //CommonHelper.savePushFileRecord(jsonFile, fileName);//存入文件推送记录
                }
               #endregion
           }      
           //此表处理完成，更新下次同步时间
           DatabaseHelper.ExecuteSql(String.Format("update sync_TableInfo st set st.lastsynctime=st.nextsynctime,st.nextsynctime= st.nextsynctime+ st.sync_interval/24/60  where st.tablename='{0}'", row["TABLENAME"].ToString()));
       }

       /// <summary>
       /// 处理推送数据,经本地测试一分钟能推送1.4W左右数据
       /// </summary>
       void DealSyncData() {
           DataTable tabDt = GetSyncTableData();
           var paraOptions = new ParallelOptions();
            paraOptions.MaxDegreeOfParallelism = maxParallel;
             
            Parallel.ForEach(tabDt.Select(), paraOptions, (row) =>
             {
                 try
                 {
                     DealSyncData(row);
                 }
                 catch (Exception ex)
                 {
                     LogHelp.WriteError(ex.Message, ex);
                 }
             });

           //foreach (DataRow row in tabDt.Rows)
           //{
           //    DataTable dt = GetSyncData(row);
           //    try
           //    {
           //        DealSyncData(dt);
           //    }
           //    catch (Exception ex)
           //    {
           //        LogHelp.WriteError(ex.Message, ex);
           //    }

           //}
       }

    }
}
